/*
 *
 */
#include "postgres_fe.h"
#include "getopt_long.h"
#include "libpq/libpq-fe.h"
#include <sys/time.h>
#include <pthread.h>
#include "gssignal/gs_signal.h"

#define CONNECT_STR_MAX_LENGTH 1024
#define SQL_MAX_LENGTH 1024
#define BATCHSQL_MAX_LENGTH (10 * 1024 * 1024)
#define TABLE_SQL_MAX_LENGTH 102400
#define INVALID_THREAD ((pthread_t)0)

#define USECS_PER_SEC 1000000
#define USECS_PER_MSEC 1000

#define RUN_KIND_NOMAL              1
#define RUN_KIND_PREPARED           2
#define RUN_KIND_PREPARED_WITHXACT  3
#define RUN_KIND_BATCHINSERT        4 //insert into values(...),(...)
#define RUN_KIND_BATCHINSERT_NOBIND 5
#define RUN_KIND_BATCHINSERT_MUTI_INSERT_NOBIND 6
#define RUN_KIND_COPY 7
#define RUN_KIND_BATCHINSERT_GAUSS 8 
#define RUN_KIND_QUERY 9
#define RUN_KIND_PREPARE_QUERY 10
#define RUN_KIND_MAX 11

#define TESTMODE_FIX_LINES 0
#define TESTMODE_FIX_TIME 1


typedef struct ThreadData
{
    pthread_t   thread_ptr;
    int         thread_id;
    int         total_round;
    PGconn      *conn;
    char        **insert_sql;

    char        batch_sql_nomal[BATCHSQL_MAX_LENGTH];
    char        batch_sql_nomal_2[BATCHSQL_MAX_LENGTH];
    char        batch_sql_endbatch[BATCHSQL_MAX_LENGTH];
    bool        insert_init;
    char        **stmt_name;
    char        **paramValues;
    char        **paramValues_2;
    int64       wait_return_time;

    char        copy_sql_buffer[BATCHSQL_MAX_LENGTH];
    
    struct timeval      start_time;
    struct timeval      stop_time;
    int64               total_time;

    struct timeval      produce_data_start_time;
    struct timeval      produce_data_stop_time;
    int64               produce_data_time;

    struct timeval      exec_start_time;
    struct timeval      exec_end_time;
    int64               exec_total_time;

    struct timeval      exec_start_dou_time;
    struct timeval      exec_end_dou_time;
    int64               exec_total_dou_time;

    struct timeval      exec_start_conn_time;
    struct timeval      exec_end_conn_time;
    int64               exec_conn_time;
    int64               prepare_total_time;


    int64         xact_num;
    int64         tuple_num;
    int64         tuple_num_fix_time;
    int         runkind;
    bool        enable_opfusion;

    bool        finish;
}ThreadData;


char    conn_str[CONNECT_STR_MAX_LENGTH] = {0};
bool    b_conn_str = false;

bool    run_work = false;
int64   xact_num = 0;
int64   tuple_num = 0;

ThreadData  *thread_handle = NULL;
THR_LOCAL ThreadData  *cur_thread = NULL;

static void* threadRun(void* arg);

//----hard code-------
char *table_name = NULL;
char *table_name_nomal = "my_test_table_nomal";
char *table_name_partition = "my_test_table_partition";
char *table_name_ts = "my_test_table_ts";

//----CONFIGURE------
int     nthread = 10;
int     total_round = 100;
int     avg_round = 10;
int     first_round = 10;
bool    do_init = false;
bool    partition_table = false;
bool    timeseries_table = false;
int     kind = 1;
int     batch_per_xact = 1;
bool    enable_opfusion = false;
bool    enable_plan_recycle = false;
bool    fast_extend = false;
bool    enable_batchinsert_to_muti = false;
bool    verbose = false;
char    relname[SQL_MAX_LENGTH];
int     target_rel_mo = 2;
bool    short_conn = false;
bool    dir_query = false;
bool    auto_test = false;
int     test_mode = TESTMODE_FIX_LINES; // 0 avg every thread  1 lines in time
uint64  run_seconds = 0;
bool    replan_every_batch = false;
int     ntables = 1;
//--connection arg
char*   hostname = NULL;
char*   conf_port = NULL;
char*   conf_user_name = NULL;
char*   conf_database = NULL;
char*   conf_password = NULL;

//------GLOBAL-------
struct timeval      start_time;
struct timeval      stop_time;
int64     real_time = 0;
int64     max_thread_time = 0;
int64     all_thread_time = 0;
int64     all_produce_time = 0;
int64     all_dou_time = 0;
int64     all_exec_time = 0;
int64     all_db_time = 0;
int64     all_conn_time = 0;
int64     all_prepare_time = 0;
bool    get_arg = false;
bool    a_thread_exit = false;
bool    conf_time_arr = false;
int64   all_tuples_fix_time = 0;


THR_LOCAL char    name_str[100] = {0};
THR_LOCAL char*   fleet[4] = {"East", "West", "South", "North"};
THR_LOCAL char*   driver[10] = {"driver1", "driver2", "driver3", "driver4", "driver5","driver6","driver7","driver8","driver9","driver10",};
THR_LOCAL char*   model[6] = {"mode1", "mode2", "mode3", "mode4", "mode5", "mode6"};
THR_LOCAL char*   device_version= "v1";
THR_LOCAL char*   load_capacity[4] = {"1000", "2000", "3000", "4000"};
THR_LOCAL char*   fuel_capacity[4] = {"300", "200", "100", "150"};
THR_LOCAL char*   nominal_fuel_consumption[4] = {"20", "20", "30", "50"};
THR_LOCAL char*   temp_value= "200.012";
THR_LOCAL int  prepare_name_loop = 0;
THR_LOCAL char    *timeinsert = NULL;

static char* get_partition_table_sql(bool pt, bool tt, char *create_table_normal, int index=0);
static void doPrepare(PGconn* conn, char *sql, int paranum,int index=0);
static void executePrepared(PGconn* conn, int argnum, int index = 0);

static void get_conn_str(void)
{
    char *host = "127.0.0.1";
    char *port = "10020";
    char *user = "lchch";
    char *dbname = "postgres";
    char *password = "123qweQWE";

    if(hostname)
        host = hostname;
    if(conf_port)
        port = conf_port;
    if(conf_user_name)
        user = conf_user_name;
    if(conf_database)
        dbname = conf_database;
    if(conf_password)
        password = conf_password;

    sprintf(conn_str, "host=%s port=%s user=%s dbname=%s password=%s ",host, port, user, dbname, password);
    b_conn_str = true;
}

static void get_conn_str1(void)
{
    char *host = "localhost";
    char *port = "10030";
    char *user = "lchch";
    char *dbname = "postgres";
    char *password = "123qweQWE";

    sprintf(conn_str, "host=%s port=%s user=%s dbname=%s password=%s ",host, port, user, dbname, password);
    b_conn_str = true;
}

static void get_conn_str2(void)
{
    char *host = "localhost";
    char *port = "10030";
    char *user = "lchch";
    char *dbname = "db1";
    char *password = "123qweQWE";

    sprintf(conn_str, "host=%s port=%s user=%s dbname=%s password=%s ",host, port, user, dbname, password);
    b_conn_str = true;
}

/* discard connections */
static void disconnect(PGconn **conn)
{
    PQfinish(*conn);
    *conn = NULL;
}

/* set up a connection to the backend */
static PGconn* doConnect(void)
{
    PGconn *conn = NULL;

    Assert(b_conn_str);
    conn = PQconnectdb(conn_str);

    if(!conn)
    {
        printf("Can not connect :!conn\n");
        exit(1);
    }

    if(conn && PQstatus(conn) == CONNECTION_BAD)
    {
        printf("Can not connect :%s\n", PQerrorMessage(conn));
        exit(1);
    }

    return conn;
}

char* get_insert_time(int count, char* target)
{
#define US_PER_SECOND (1000000L)
#define US_PER_MINUTE (1000000L * 60)
#define US_PER_HOUR (1000000L * 60 * 60)
#define US_PER_DAY (1000000L * 60 * 60 * 24)
    int64   time_offset;
    int     day = 1;
    int     hour = 0;
    int     minute = 0;
    int     second = 0;
    int     usecond = 0;
    struct timeval   time_now;
    
    gettimeofday(&time_now, NULL);

    if(!target)
    {
        target = (char*)malloc(30);
    }
    memset(target, 0, 30);

    time_offset = (int64)count * US_PER_SECOND;
    usecond = time_now.tv_usec % US_PER_SECOND;
    second = (int)((time_offset % US_PER_MINUTE) / US_PER_SECOND);
    minute = (int)((time_offset % US_PER_HOUR) / US_PER_MINUTE);
    hour = (int)((time_offset % US_PER_DAY) / US_PER_HOUR);
    day += (int)((time_offset / US_PER_DAY));

    sprintf(target, "2023-8-%d %d:%d:%d.%d", day, hour, minute, second, usecond);
    return target;
}

static void executeCommand(PGconn* conn, const char* sql)
{
    PGresult* res = NULL;

    res = PQexec(conn, sql);
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        disconnect(&conn);
        exit(1);
    }
    PQclear(res);
}

static PGresult* executeQuery(PGconn* conn, const char* sql, bool permit_error)
{
    PGresult* res = NULL;

    res = PQexec(conn, sql);
    if (PQresultStatus(res) != PGRES_TUPLES_OK && !permit_error) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        disconnect(&conn);
        exit(1);
    }
    return res;
}

static void executeBegin(PGconn* conn)
{
    PGresult    *res = NULL;
    char        *sql = "begin;";

    res = PQexec(conn, sql);
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        disconnect(&conn);
        exit(1);
    }
    PQclear(res);
}

static void executeCommit(PGconn* conn)
{
    PGresult    *res = NULL;
    char        *sql = "commit;";

    res = PQexec(conn, sql);
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        disconnect(&conn);
        exit(1);
    }
    PQclear(res);
}

static void executeDropstmt(PGconn* conn, bool other)
{
    PGresult    *res = NULL;
    char        sqBUf[100] = {0};
    char        *sql = NULL;
    char* stmt_name = NULL;

    if(other)
        stmt_name = cur_thread->stmt_name[1];
    else
        stmt_name = cur_thread->stmt_name[0];

    sprintf(sqBUf, "deallocate \"%s\"", stmt_name);
    sql = sqBUf;
//printf("%s\n", sql);
    res = PQexec(conn, sql);
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        disconnect(&conn);
        exit(1);
    }
    PQclear(res);
}


static void doPrepare(PGconn* conn, char *sql, int paranum,  int index)
{
    PGresult*   res = NULL;
    char*       stmt_name = NULL;

    stmt_name = cur_thread->stmt_name[index];

    res = PQprepare(conn, stmt_name, sql, paranum, NULL);
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        disconnect(&conn);
        exit(1);
    }
    PQclear(res);
}

static void executePrepared(PGconn* conn, int argnum, int index)
{
    PGresult    *res = NULL;

    char*       stmt_name = NULL;
    char**      paramValues = NULL;

    Assert(ntables > index);

    /*if(index != 0) {
        paramValues = cur_thread->paramValues_2;
    } else */
    {
        paramValues = cur_thread->paramValues;
    }
    stmt_name = cur_thread->stmt_name[index];

    res = PQexecPrepared(conn,
            stmt_name,
            argnum,
            paramValues,
            NULL,
            NULL,
            1
            );
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        disconnect(&conn);
        
        exit(1);
    }
    PQclear(res);
}

static void doPrepare_query(PGconn* conn, char *sql)
{
    PGresult* res = NULL;

    res = PQprepare(conn, cur_thread->stmt_name[0], sql, 0, NULL);
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        //disconnect(&conn);
        //exit(1);
    }
    else
        PQclear(res);
}

static void executePrepared_query(PGconn* conn, int argnum)
{
    PGresult    *res = NULL;

    res = PQexecPrepared(conn,
            cur_thread->stmt_name[0],
            argnum,
            cur_thread->paramValues,
            NULL,
            NULL,
            1
            );

    if (PQresultStatus(res) != PGRES_TUPLES_OK) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        //disconnect(&conn);
        //exit(1);
    }
    else
        PQclear(res);
}

static void executeBatchinsertGauss(PGconn* conn, int param, int tuple_num)
{
    PGresult    *res = NULL;

    res = PQexecParamsBatch(conn,
          cur_thread->batch_sql_nomal,
          param,
          tuple_num,
          NULL,
          cur_thread->paramValues,
          NULL,
          NULL,
          1);
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        disconnect(&conn);
        
        exit(1);
    }
    PQclear(res);
}
/*
static void set_enable_opfusion(PGconn *conn, bool enable)
{
    char set_sql[SQL_MAX_LENGTH] = {0};
    PGresult*   res = NULL;
    char*       guc_name = NULL;
    char*       guc_name1 = "enable_opfusion";
    char*       guc_name2 = "bypass_for_prepared_values_batch";

    if(kind == RUN_KIND_BATCHINSERT)
        guc_name = guc_name2;
    else
        guc_name = guc_name1;

    if(enable) {
            sprintf(set_sql, "set %s=on", guc_name);
    } else {
        sprintf(set_sql, "set %s=off", guc_name);
    }
    res = executeQuery(conn, set_sql, true);

    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        disconnect(&conn);
        fprintf(stderr, "set enable_opfusion failed\n");
        exit(1);
    }
    PQclear(res);
}

static void set_enable_plan_recycle(PGconn *conn, bool enable)
{
    char set_sql[SQL_MAX_LENGTH] = {0};
    PGresult*   res = NULL;
    char*       guc_name = "uqbar.enable_plan_recycle";


    if(enable) {
            sprintf(set_sql, "set %s=on", guc_name);
    } else {
        sprintf(set_sql, "set %s=off", guc_name);
    }
    res = executeQuery(conn, set_sql, true);

    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        disconnect(&conn);
        fprintf(stderr, "set enable_plan_recycle failed\n");
        exit(1);
    }
    PQclear(res);
}
*/
/*
static void set_enable_batchinsert_to_muti(PGconn *conn, bool enable)
{
    char set_sql[SQL_MAX_LENGTH] = {0};
    PGresult    *res = NULL;

    if(!enable)
        return;

    if(enable)
        sprintf(set_sql, "set mutiinser_for_prepared_values_batch=on");
    else
        sprintf(set_sql, "set mutiinser_for_prepared_values_batch=off");

    res = executeQuery(conn, set_sql, true);

    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        disconnect(&conn);
        fprintf(stderr, "set mutiinser_for_prepared_values_batch failed\n");
        exit(1);
    }
    PQclear(res);
}
*/
/*
static void set_enable_fast_extend(PGconn *conn, bool enable)
{
    char set_sql[SQL_MAX_LENGTH] = {0};
    PGresult    *res = NULL;

    if(!enable)
        return;

    if(enable)
        sprintf(set_sql, "set fast_extend_for_prepared_values_batch=on");
    else
        sprintf(set_sql, "set fast_extend_for_prepared_values_batch=off");

    res = executeQuery(conn, set_sql, true);

    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        disconnect(&conn);
        fprintf(stderr, "set fast_extend_for_prepared_values_batch failed\n");
        exit(1);
    }
    PQclear(res);
}
*/
static void check_table_exists(PGconn *conn, char* the_name)
{
    char        check_sql[SQL_MAX_LENGTH] = {0};
    PGresult    *res = NULL;

    sprintf(check_sql, "select * from %s limit 1", the_name);
    res = executeQuery(conn, check_sql, true);

    if (PQresultStatus(res) != PGRES_TUPLES_OK) {
        PQclear(res);
        disconnect(&conn);
        fprintf(stderr, "Do test table exists?\n");
        exit(1);
    }
    PQclear(res);
}

static int64 get_time_diff_us(struct timeval start, struct timeval end)
{
    int64   microsecond = 0;

    microsecond = (end.tv_sec * USECS_PER_SEC + end.tv_usec) - (start.tv_sec * USECS_PER_SEC + start.tv_usec);
    return microsecond;
}

void do_query(PGconn *conn)
{
    PGresult* res = NULL;
    char* count = 0;
    char    sql[SQL_MAX_LENGTH] = {0};
    memset(sql, 0, SQL_MAX_LENGTH);
    sprintf(sql, "SELECT count(*) FROM my_test_table_nomal");

    res = executeQuery(conn, sql, false);
    if(PQresultStatus(res) != PGRES_TUPLES_OK)
    {
        printf("[handle_query]PQresultStatus(res) != PGRES_TUPLES_OK");
        exit(1);
    }
    count = PQgetvalue(res, 0,0);
    printf("COUNT:%s\n", count);
}

void do_query_prepare(PGconn *conn)
{
    char    sql[SQL_MAX_LENGTH] = {0};
    static bool inited = false;
    static int name_count = 0;

    memset(sql, 0, SQL_MAX_LENGTH);
    sprintf(sql, "SELECT count(*) FROM t1");

    sprintf(cur_thread->stmt_name[0], "my_test_tool_stmt_%d_%d", cur_thread->thread_id, name_count++);
printf("aaaa %s\n", sql);
    if(!inited) {
        doPrepare_query(conn, sql);
printf("aabb\n");
        inited = false;
    }
printf("bbb\n");
    executePrepared_query(conn, 0);
printf("ccc\n");
sleep(2);
}

void handle_query(void)
{

    PGconn *conn = doConnect();
    char    sql[SQL_MAX_LENGTH] = {0};
    int     loop = 0;
    char    *count = NULL;

    while(true)
    {
        PGresult* res = NULL;
        memset(sql, 0, SQL_MAX_LENGTH);
        sprintf(sql, "SELECT count(*) FROM heap_page_items(get_raw_page('my_test_table_nomal', %d))", loop++);

        res = executeQuery(conn, sql, false);
        if(PQresultStatus(res) != PGRES_TUPLES_OK)
        {
            printf("[handle_query]PQresultStatus(res) != PGRES_TUPLES_OK");
            exit(1);
        }
        count = PQgetvalue(res, 0,0);
        if(atoi(count) < 156)
            printf("PAGE:%d,COUNT:%s\n", loop - 1, count);

    }
}

void test_ts_feature(char* sql)
{
    PGconn      *conn = doConnect();
    char        *stmt_name = "test_stmt";
    PGresult    *res = NULL;

    printf("testsql=%s\n", sql);
    fflush(stdout);

    res = PQprepare(conn, stmt_name, sql, 0, NULL);
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        disconnect(&conn);
        exit(1);
    }
    PQclear(res);

    res = PQexecPrepared(conn,
            stmt_name,
            0,
            NULL,
            NULL,
            NULL,
            1
            );
    if (PQresultStatus(res) != PGRES_COMMAND_OK) {
        PQclear(res);
        fprintf(stderr, "%s", PQerrorMessage(conn));
        disconnect(&conn);
        
        exit(1);
    }
    PQclear(res);
    disconnect(&conn);
}

void handle_auto_test(void)
{
    //get_conn_str1();   
    
    //handle_query();
    ;

}

static bool do_insert_prepare(PGconn *conn, int round, int total_round)
{
    int     loop = 0;
    int     id = rand() % 100;
    int     profit = 1000 + rand() % 1000;
    char    *randomstr = "123qweQWE123";
    static THR_LOCAL char    *timeinsert = NULL;

    char    id_str[10] = {0};
    char    profit_str[10] = {0};

    sprintf(id_str, "%d", id);
    sprintf(profit_str, "%d", profit);
    timeinsert = get_insert_time(round,timeinsert);    
    if(!cur_thread->insert_init)
    {
        for(; loop < ntables; loop++)
        {
            sprintf(cur_thread->insert_sql[loop], "INSERT INTO %s_%d VALUES($1,$2,$3,$4)", table_name, loop);
            sprintf(cur_thread->stmt_name[loop], "my_test_tool_stmt_%d_%d", loop, cur_thread->thread_id);
            doPrepare(conn, cur_thread->insert_sql[loop], 4, loop);
        }
        cur_thread->paramValues = (char**)malloc(sizeof(char*) * 4);
        cur_thread->insert_init = true;
    }

    memset(cur_thread->paramValues, 0, sizeof(char*) * 4);
    cur_thread->paramValues[0] = id_str;
    cur_thread->paramValues[1] = profit_str;
    cur_thread->paramValues[2] = randomstr;
    cur_thread->paramValues[3] = (char*)timeinsert;

    if(cur_thread->runkind == RUN_KIND_PREPARED_WITHXACT && (round % batch_per_xact == 0) )
    {
        gettimeofday(&cur_thread->exec_start_time, NULL);
        executeBegin(conn);
        gettimeofday(&cur_thread->exec_end_time, NULL);
        cur_thread->exec_total_time += get_time_diff_us(cur_thread->exec_start_time, cur_thread->exec_end_time);

    }

    gettimeofday(&cur_thread->exec_start_time, NULL);
    for(loop=0; loop < ntables; loop++)
    {
        executePrepared(conn, 4, loop);
    }
    gettimeofday(&cur_thread->exec_end_time, NULL);
    cur_thread->exec_total_time += get_time_diff_us(cur_thread->exec_start_time, cur_thread->exec_end_time);
    cur_thread->tuple_num += ntables;

    if(cur_thread->runkind == RUN_KIND_PREPARED_WITHXACT && (round % batch_per_xact == batch_per_xact - 1 || round == total_round - 1) )
    {
        cur_thread->xact_num++;
        gettimeofday(&cur_thread->exec_start_time, NULL);
        executeCommit(conn);
        gettimeofday(&cur_thread->exec_end_time, NULL);
        cur_thread->exec_total_time += get_time_diff_us(cur_thread->exec_start_time, cur_thread->exec_end_time);
    }
    return true;
}

static bool do_insert_batch_nobind(PGconn *conn, int round, int total_round)
{
    int     id = rand() % 100;
    int     profit = 1000 + rand() % 1000;
    char    *randomstr = "123qweQWE123";

    char    id_str[10] = {0};
    char    profit_str[10] = {0};

    char    *sql_ptr = NULL;
    static THR_LOCAL char    *timeinsert = NULL;

    sprintf(id_str, "%d", id);
    sprintf(profit_str, "%d", profit);
    timeinsert = get_insert_time(round,timeinsert); 
    if(!cur_thread->insert_init)
    {
        memset(cur_thread->batch_sql_endbatch, 0, BATCHSQL_MAX_LENGTH);
        cur_thread->insert_init = true;
    }


    sql_ptr = cur_thread->batch_sql_endbatch + strlen(cur_thread->batch_sql_endbatch);

    if((round % batch_per_xact == 0) )
    {

        sprintf(sql_ptr, "INSERT INTO %s VALUES(%s,%s,\'%s\', \'%s\')",
                            table_name, id_str, profit_str, randomstr, timeinsert);
    }
    else
    {
        if(sql_ptr - cur_thread->batch_sql_endbatch >= BATCHSQL_MAX_LENGTH - 1000)
        {
            printf("[do_insert_batch][threadid=%d] SQL length limit: %ld", cur_thread->thread_id, sql_ptr - cur_thread->batch_sql_endbatch);
            a_thread_exit = true;
            exit(1);
        }
        sprintf(sql_ptr, ", (%s,%s,\'%s\', \'%s\')",
                            id_str, profit_str, randomstr, timeinsert);
    }

    if((round % batch_per_xact == batch_per_xact - 1) || round == total_round - 1)
    {
        sql_ptr = cur_thread->batch_sql_endbatch + strlen(cur_thread->batch_sql_endbatch);
        sprintf(sql_ptr, ";");
//printf("[do_insert_batch]cur_thread->batch_sql_endbatch=%s\n", cur_thread->batch_sql_endbatch);
        gettimeofday(&cur_thread->exec_start_time, NULL);
        executeCommand(conn, cur_thread->batch_sql_endbatch);
        gettimeofday(&cur_thread->exec_end_time, NULL);
        cur_thread->exec_total_time += get_time_diff_us(cur_thread->exec_start_time, cur_thread->exec_end_time);
        if(short_conn)
            cur_thread->insert_init = false;
        return true;
    }
    return false;
}

//executeCommand(cur_conn, drop_table);
//executeCommand(cur_conn, create_table);
static bool do_copy_batch(PGconn *conn, int round, int total_round)
{
    int     id = rand() % 100;
    int     profit = 1000 + rand() % 1000;
    char    *randomstr = "123qweQWE123";
    char    *sql_ptr = NULL;
    static THR_LOCAL char    *timeinsert = NULL;
    char    test_copy_sql[1024] = {0};
//printf("do_copy_batch %d\n", round);

    if(!cur_thread->insert_init)
    {
        memset(cur_thread->copy_sql_buffer, 0, BATCHSQL_MAX_LENGTH);
        cur_thread->insert_init = true;
    }
    timeinsert = get_insert_time(round,timeinsert); 
    sql_ptr = cur_thread->copy_sql_buffer + strlen(cur_thread->copy_sql_buffer);
    sprintf(sql_ptr, "%d\t%d\t%s\t%s\n", id, profit, randomstr, timeinsert);
//printf("do_copy_batch  %d\n", round);
    if((round % batch_per_xact == batch_per_xact - 1) || round == total_round - 1)
    {
        PGresult  *result_status;
        sprintf(test_copy_sql, "COPY %s FROM stdin;", table_name);
        gettimeofday(&cur_thread->exec_start_time, NULL);
//printf("do_copy_batch--- %d\n", round);
        result_status = PQexec(conn, test_copy_sql);
        assert(PGRES_COPY_IN == PQresultStatus(result_status));
        if(PQputCopyData(conn, cur_thread->copy_sql_buffer, strlen(cur_thread->copy_sql_buffer)) <= 0
            || PQflush(conn))
        {
            printf("Can not send copy:%s\n", PQerrorMessage(conn));
            exit(1);
        }

        if(PQputCopyEnd(conn, NULL) < 0 || PQflush(conn))
        {
            printf("Can not end copy:%s\n", PQerrorMessage(conn));
            exit(1);
        }
        gettimeofday(&cur_thread->exec_end_time, NULL);
        cur_thread->exec_total_time += get_time_diff_us(cur_thread->exec_start_time, cur_thread->exec_end_time);
        memset(cur_thread->copy_sql_buffer, 0, BATCHSQL_MAX_LENGTH);
        return true;
    }
    return false;
}

static bool do_mutiinsert_batch_nobind(PGconn *conn, int round, int total_round)
{
    int     id = rand() % 100;
    int     profit = 1000 + rand() % 1000;
    char    *randomstr = "123qweQWE123";
    char    *sql_ptr = NULL;
    static THR_LOCAL char    *timeinsert = NULL;


    if(!cur_thread->insert_init)
    {
        memset(cur_thread->batch_sql_nomal, 0, BATCHSQL_MAX_LENGTH);
        cur_thread->insert_init = true;
    }

    sql_ptr = cur_thread->batch_sql_nomal + strlen(cur_thread->batch_sql_nomal);
    if(sql_ptr - cur_thread->batch_sql_endbatch >= BATCHSQL_MAX_LENGTH - 100)
    {
        printf("[do_mutiinsert_batch][threadid=%d] SQL length limit：%ld", cur_thread->thread_id, sql_ptr - cur_thread->batch_sql_endbatch);
        a_thread_exit = true;
        exit(1);
    }

    timeinsert = get_insert_time(round,timeinsert); 
    sprintf(sql_ptr, "INSERT INTO %s VALUES(%d,%d,\'%s\', \'%s\');", table_name, id, profit, randomstr, timeinsert);

    if((round % batch_per_xact == batch_per_xact - 1 || round == total_round - 1) )
    {
        gettimeofday(&cur_thread->exec_start_time, NULL);
        executeCommand(conn, cur_thread->batch_sql_nomal);
        gettimeofday(&cur_thread->exec_end_time, NULL);
        cur_thread->exec_total_time += get_time_diff_us(cur_thread->exec_start_time, cur_thread->exec_end_time);
        memset(cur_thread->batch_sql_nomal, 0, BATCHSQL_MAX_LENGTH);
        
        if(short_conn)
            cur_thread->insert_init = false;
        return true;
    }
    return false;

}

static bool do_insert_batch_gauss(PGconn *conn, int round, int total_round)
{
    int     id = rand() % 100;
    int     profit = 1000 + rand() % 1000;
    char    *randomstr = "123qweQWE123";
    static THR_LOCAL char    *timeinsert = NULL;

    char    id_str[10] = {0};
    char    profit_str[10] = {0};

    sprintf(id_str, "%d", id);
    sprintf(profit_str, "%d", profit);
    timeinsert = get_insert_time(round,timeinsert); 
    if(!cur_thread->insert_init)
    {
        memset(cur_thread->batch_sql_nomal, 0, BATCHSQL_MAX_LENGTH);
        memset(cur_thread->batch_sql_endbatch, 0, BATCHSQL_MAX_LENGTH);
        sprintf(cur_thread->batch_sql_nomal, "INSERT INTO %s VALUES($1, $2, $3, $4)", table_name);
        cur_thread->paramValues = (char**)malloc(sizeof(char*) * 4 * batch_per_xact);
        memset(cur_thread->paramValues, 0, sizeof(char*) * 4 * batch_per_xact);
        cur_thread->insert_init = true;
    }

    {
        int itera = round % batch_per_xact;
        static THR_LOCAL int batch_nums = 0;

        if((round % batch_per_xact == 0) )
        {
            memset(cur_thread->paramValues, 0, sizeof(char*) * 4 * batch_per_xact);
        }

        cur_thread->paramValues[itera * 4] = strdup(id_str);
        cur_thread->paramValues[itera * 4 + 1] = strdup(profit_str);
        cur_thread->paramValues[itera * 4 + 2] = strdup(randomstr);
        cur_thread->paramValues[itera * 4 + 3] = strdup(timeinsert);
        cur_thread->tuple_num++;
        batch_nums++;

        if((round % batch_per_xact == batch_per_xact - 1 || round == total_round - 1) )
        {
            int loop = 0;

            gettimeofday(&cur_thread->exec_start_time, NULL);
            executeBatchinsertGauss(conn, 4, batch_nums);
            gettimeofday(&cur_thread->exec_end_time, NULL);
            cur_thread->exec_total_time += get_time_diff_us(cur_thread->exec_start_time, cur_thread->exec_end_time);
            cur_thread->xact_num++;
            for(; loop < batch_per_xact; loop++)
            {
                if(cur_thread->paramValues[loop])
                    free(cur_thread->paramValues[loop]);
            }
            batch_nums = 0;

            if(short_conn)
                cur_thread->insert_init = false;
            return true;
        }
    }
    return false;
}

static bool do_insert_batch(PGconn *conn, int round, int total_round)
{
gettimeofday(&cur_thread->exec_start_dou_time, NULL);
    int     id = round % 5000;
    int     profit = 1000 + id % 1000;
gettimeofday(&cur_thread->exec_end_dou_time, NULL);
cur_thread->exec_total_dou_time += get_time_diff_us(cur_thread->exec_start_dou_time, cur_thread->exec_end_dou_time);
    int     columns = 16;
    char    id_str[10] = {0};
    char    profit_str[10] = {0};
    int     nomal_batchs = 0;
    bool    last_batch = false;
    static THR_LOCAL bool    need_replan = false;
    static THR_LOCAL int     batch_per_xact_temp = batch_per_xact;
    static THR_LOCAL int     cur_irate = 0;

    if(total_round != -1) {
        nomal_batchs = total_round / batch_per_xact_temp;
        last_batch = ((round / batch_per_xact_temp) == nomal_batchs && total_round % batch_per_xact_temp != 0);
    }
    last_batch = false;

    sprintf(id_str, "%d", id);
    sprintf(name_str, "truck_%d", id);
    sprintf(profit_str, "%d", profit);

    timeinsert = get_insert_time(round,timeinsert);

    if(!cur_thread->insert_init || need_replan)
    {
        int     loop = 0;
        int     loop_i = 0;
        char    *sql_ptr = NULL;
        memset(cur_thread->batch_sql_nomal, 0, BATCHSQL_MAX_LENGTH);
        memset(cur_thread->batch_sql_endbatch, 0, BATCHSQL_MAX_LENGTH);
        sprintf(cur_thread->batch_sql_nomal, "INSERT INTO %s_0 VALUES", table_name);

        for(; loop < batch_per_xact_temp; loop++)
        {
            sql_ptr = cur_thread->batch_sql_nomal + strlen(cur_thread->batch_sql_nomal);
            sprintf(sql_ptr, "($%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d)",
            loop * columns + 1, loop * columns + 2, loop * columns + 3, loop * columns + 4,
            loop * columns + 5, loop * columns + 6, loop * columns + 7, loop * columns + 8,
            loop * columns + 9, loop * columns + 10, loop * columns + 11, loop * columns + 12,
            loop * columns + 13, loop * columns + 14, loop * columns + 15, loop * columns + 16);
            sql_ptr = cur_thread->batch_sql_nomal + strlen(cur_thread->batch_sql_nomal);
            if(loop == batch_per_xact_temp - 1)
            {
                sprintf(sql_ptr, ";");
            }
            else
            {
                sprintf(sql_ptr, ",");
            }
            if(sql_ptr - cur_thread->batch_sql_nomal >= BATCHSQL_MAX_LENGTH - 10)
            {
                printf("[do_insert_batch][threadid=%d] SQL length limit: %ld",
                        cur_thread->thread_id, sql_ptr - cur_thread->batch_sql_nomal);
                a_thread_exit = true;
                exit(1);
            }
        }
        //printf("NOW batch_per_xact_temp=%d %s\n", batch_per_xact_temp, cur_thread->batch_sql_nomal);
        cur_thread->paramValues = (char**)malloc(sizeof(char*) * columns * batch_per_xact_temp);
        memset(cur_thread->paramValues, 0, sizeof(char*) * columns * batch_per_xact_temp);
        for(loop_i=0; loop_i < ntables; loop_i++)
        {
            memset(cur_thread->stmt_name[loop_i], 0, NAMEDATALEN);
            sprintf(cur_thread->stmt_name[loop_i], "my_test_tool_stmt_%d_%d_%d_1",
                            cur_thread->thread_id, round / batch_per_xact_temp, prepare_name_loop++);
            /*sprintf(cur_thread->stmt_name[loop_i], "my_test_tool_stmt_%d_1",
                            cur_thread->thread_id);*/
        }
    }

    if((replan_every_batch && need_replan) || !cur_thread->insert_init) {
        gettimeofday(&cur_thread->exec_start_time, NULL);
        doPrepare(conn, cur_thread->batch_sql_nomal, batch_per_xact_temp * columns);
        gettimeofday(&cur_thread->exec_end_time, NULL);
        cur_thread->prepare_total_time += get_time_diff_us(cur_thread->exec_start_time, cur_thread->exec_end_time);
        cur_thread->insert_init = true;
        need_replan = false;
    }

    if(!last_batch) {
        int itera = cur_irate;

        gettimeofday(&cur_thread->produce_data_start_time, NULL);
        if(itera == 0)
        {
            memset(cur_thread->paramValues, 0, sizeof(char*) * columns * batch_per_xact_temp);
        }

        cur_thread->paramValues[itera * columns] = strdup(timeinsert);
        cur_thread->paramValues[itera * columns + 1] = strdup(name_str);
        cur_thread->paramValues[itera * columns + 2] = strdup(fleet[id%4]);
        cur_thread->paramValues[itera * columns + 3] = strdup(driver[id%10]);
        cur_thread->paramValues[itera * columns + 4] = strdup(model[id%6]);
        cur_thread->paramValues[itera * columns + 5] = strdup(device_version);
        cur_thread->paramValues[itera * columns + 6] = strdup(load_capacity[id%4]);
        cur_thread->paramValues[itera * columns + 7] = strdup(fuel_capacity[id%4]);
        cur_thread->paramValues[itera * columns + 8] = strdup(nominal_fuel_consumption[id%4]);
        cur_thread->paramValues[itera * columns + 9] = strdup(temp_value);
        cur_thread->paramValues[itera * columns + 10] = strdup(temp_value);
        cur_thread->paramValues[itera * columns + 11] = strdup(temp_value);
        cur_thread->paramValues[itera * columns + 12] = strdup(temp_value);
        cur_thread->paramValues[itera * columns + 13] = strdup(temp_value);
        cur_thread->paramValues[itera * columns + 14] = strdup(temp_value);
        cur_thread->paramValues[itera * columns + 15] = strdup(temp_value);
        cur_thread->tuple_num++;
        cur_irate++;

        gettimeofday(&cur_thread->produce_data_stop_time, NULL);
        cur_thread->produce_data_time += get_time_diff_us(cur_thread->produce_data_start_time, cur_thread->produce_data_stop_time);

        if((cur_irate == batch_per_xact_temp || round == total_round - 1) )
        {
            int loop = 0;

//printf("cur_irate=%d round=%d batch_per_xact_temp=%d\n", cur_irate, round, batch_per_xact_temp);
            gettimeofday(&cur_thread->exec_start_time, NULL);
            executePrepared(conn, batch_per_xact_temp * columns);
            gettimeofday(&cur_thread->exec_end_time, NULL);
            cur_thread->exec_total_time += get_time_diff_us(cur_thread->exec_start_time, cur_thread->exec_end_time);
            cur_thread->xact_num++;
            gettimeofday(&cur_thread->produce_data_start_time, NULL);
            for(; loop < batch_per_xact_temp; loop++)
            {
                if(cur_thread->paramValues[loop])
                    free(cur_thread->paramValues[loop]);
            }
            gettimeofday(&cur_thread->produce_data_stop_time, NULL);
            cur_thread->produce_data_time += get_time_diff_us(cur_thread->produce_data_start_time, cur_thread->produce_data_stop_time);

            if(replan_every_batch) {
                executeDropstmt(conn, false);
                need_replan = true;
                batch_per_xact_temp = batch_per_xact - cur_thread->produce_data_stop_time.tv_usec % 5;
                if(batch_per_xact_temp < 0)
                    batch_per_xact_temp = 1;
            }

            if(short_conn) {
                cur_thread->insert_init = false;
            }
            cur_irate = 0;
            return true;
        }
    }

    return false;
}

static bool do_insert_nomal(PGconn *conn, int round)
{
    int     id = rand() % 100;
    int     profit = 1000 + rand() % 1000;
    char    *randomstr = "123qweQWE123";
    int     loop = 0;

    char    id_str[10] = {0};
    char    profit_str[10] = {0};
    static THR_LOCAL char    *timeinsert = NULL;

    sprintf(id_str, "%d", id);
    sprintf(profit_str, "%d", profit);
    
    timeinsert = get_insert_time(round,timeinsert); 

    for(; loop < ntables; loop++)
    {
        memset(cur_thread->insert_sql[0], 0, SQL_MAX_LENGTH);
        sprintf(cur_thread->insert_sql[0], "INSERT INTO %s_%d VALUES(%s,%s,\'%s\', \'%s\')",
                                table_name, loop, id_str, profit_str, randomstr, timeinsert);
        gettimeofday(&cur_thread->exec_start_time, NULL);
        executeCommand(conn, cur_thread->insert_sql[0]);
        gettimeofday(&cur_thread->exec_end_time, NULL);
    }
    cur_thread->exec_total_time += get_time_diff_us(cur_thread->exec_start_time, cur_thread->exec_end_time);
    cur_thread->xact_num += ntables;
    cur_thread->tuple_num += ntables;
    return true;
}

static void* threadRun(void* arg)
{
    int         round = 0;
    bool        run_insert = false;

    cur_thread = (ThreadData*)arg;
    while(!run_work) {
        sleep(1);
    }

    /*set_enable_opfusion(cur_thread->conn, enable_opfusion);
    set_enable_batchinsert_to_muti(cur_thread->conn, enable_batchinsert_to_muti);
    set_enable_fast_extend(cur_thread->conn, fast_extend);
    set_enable_plan_recycle(cur_thread->conn, enable_plan_recycle);*/

    gettimeofday(&cur_thread->start_time, NULL);
    for(; round < cur_thread->total_round || test_mode == TESTMODE_FIX_TIME; round++)
    {
        if(cur_thread->runkind == RUN_KIND_NOMAL)
            run_insert = do_insert_nomal(cur_thread->conn, round);
        else if(cur_thread->runkind == RUN_KIND_PREPARED_WITHXACT || cur_thread->runkind == RUN_KIND_PREPARED) {
            run_insert = do_insert_prepare(cur_thread->conn, round, cur_thread->total_round);
        }
        else if(cur_thread->runkind == RUN_KIND_BATCHINSERT) {
            run_insert = do_insert_batch(cur_thread->conn, round, cur_thread->total_round);
        }
        else if(cur_thread->runkind == RUN_KIND_BATCHINSERT_NOBIND)
            run_insert = do_insert_batch_nobind(cur_thread->conn, round, cur_thread->total_round);
        else if(cur_thread->runkind == RUN_KIND_BATCHINSERT_MUTI_INSERT_NOBIND) {
            run_insert = do_mutiinsert_batch_nobind(cur_thread->conn, round, cur_thread->total_round);
        }
        else if(cur_thread->runkind == RUN_KIND_BATCHINSERT_GAUSS) {
            run_insert = do_insert_batch_gauss(cur_thread->conn, round, cur_thread->total_round);
        }
        else if(cur_thread->runkind == RUN_KIND_COPY) {
            run_insert = do_copy_batch(cur_thread->conn, round, cur_thread->total_round);
        }
        else if(cur_thread->runkind == RUN_KIND_QUERY) {
            do_query(cur_thread->conn);
        }
        else if(cur_thread->runkind == RUN_KIND_PREPARE_QUERY) {
            do_query_prepare(cur_thread->conn);
        }
        if(run_insert) {
            cur_thread->tuple_num_fix_time += batch_per_xact;
        }

        if(run_insert && conf_time_arr)
            break;
    
        //在do_insert_batch和do_insert_batch模式下,在批次短连接时，需要重置conn
        if((cur_thread->runkind == RUN_KIND_BATCHINSERT || cur_thread->runkind == RUN_KIND_BATCHINSERT_NOBIND ||
            cur_thread->runkind == RUN_KIND_BATCHINSERT_GAUSS || 
            cur_thread->runkind == RUN_KIND_BATCHINSERT_MUTI_INSERT_NOBIND || cur_thread->runkind == RUN_KIND_COPY) && 
                !cur_thread->insert_init && short_conn)
        {
            disconnect(&cur_thread->conn);
            gettimeofday(&cur_thread->exec_start_conn_time, NULL);
            cur_thread->conn = doConnect();
            gettimeofday(&cur_thread->exec_end_conn_time, NULL);
            cur_thread->exec_conn_time += get_time_diff_us(cur_thread->exec_start_conn_time, cur_thread->exec_end_conn_time);
        }
    }
    gettimeofday(&cur_thread->stop_time, NULL);
    cur_thread->total_time = get_time_diff_us(cur_thread->start_time, cur_thread->stop_time);

    cur_thread->finish = true;
    return NULL;
}

static void init_thread(void)
{
    int loop = 0;
    int loop_table = 0;

    thread_handle = (ThreadData*)malloc(sizeof(ThreadData) * nthread);
    memset(thread_handle, 0, sizeof(ThreadData) * nthread);

    for (loop = 0; loop < nthread; loop++) {
        ThreadData *th_ptr = &thread_handle[loop];
        int err = pthread_create(&th_ptr->thread_ptr, NULL, threadRun, th_ptr);

        if(loop == 0)
            th_ptr->total_round = first_round;
        else
            th_ptr->total_round = avg_round;

        if(test_mode == TESTMODE_FIX_TIME)
        {
            th_ptr->total_round = total_round;
        }


        th_ptr->insert_sql = (char**)malloc(sizeof(char*) * ntables);
        th_ptr->stmt_name = (char**)malloc(sizeof(char*) * ntables);
        for(loop_table=0; loop_table < ntables; loop_table++)
        {
            th_ptr->insert_sql[loop_table] = (char*)malloc(SQL_MAX_LENGTH);
            memset(th_ptr->insert_sql[loop_table], 0, SQL_MAX_LENGTH);
            th_ptr->stmt_name[loop_table] = (char*)malloc(NAMEDATALEN);
            memset(th_ptr->stmt_name[loop_table], 0, NAMEDATALEN);
        }

        th_ptr->conn = doConnect();
        th_ptr->thread_id = loop;
        th_ptr->runkind = kind;
        th_ptr->enable_opfusion = enable_opfusion;

        if (err != 0 || th_ptr->thread_ptr == INVALID_THREAD) {
            fprintf(stderr, "cannot create thread: %s\n", strerror(err));
            exit(1);
        }
    }
}

static void
notice_down(void)
{
    int loop = 0;
    for (loop = 0; loop < nthread; loop++)
    {
        if(0 == pthread_kill(thread_handle[loop].thread_ptr, SIGINT))
        {
            printf("[notice_down]kill %d\n", thread_handle[loop].thread_id);
        }
    }
    exit(1);
}

static void
count_run_time(void)
{
    uint64  run_time_now = 0;
    if(test_mode != TESTMODE_FIX_TIME) 
    {
        return;
    }
    while(true)
    {
        gettimeofday(&stop_time, NULL);
        run_time_now = get_time_diff_us(start_time, stop_time);
        if(run_time_now / USECS_PER_SEC > run_seconds) 
        {
            conf_time_arr = true;
            break;
        }
        usleep(100L);
    }
}

static void
join_thread(void)
{
    int loop = 0;
    int finish_num = 0;

    while(finish_num < nthread)
    {
        finish_num = 0;

        for (loop = 0; loop < nthread; loop++)
        {

            if(thread_handle[loop].finish)
            {
                finish_num++;
            }
        }

        if(a_thread_exit)
        {
            notice_down();
        }
        usleep(50000);//50ms
    }

    for (loop = 0; loop < nthread; loop++)
    {
        xact_num += thread_handle[loop].xact_num;
        tuple_num += thread_handle[loop].tuple_num;
    }

    //a_thread_exit
    /*for (loop = 0; loop < nthread; loop++) {

        pthread_join(thread_handle[loop].thread_ptr, NULL);

        xact_num += thread_handle[loop].xact_num;
        tuple_num += thread_handle[loop].tuple_num;
    }*/
}

static void dump_thread(void)
{
    int loop = 0;
    for (loop = 0; loop < nthread; loop++) {
        ThreadData *thr_loop = &thread_handle[loop];

        {

            printf("Thread %d:\nTotal time %lds %ldms\nExec time %lds %ldms\n",thr_loop->thread_id, 
                        thr_loop->total_time / USECS_PER_SEC, (thr_loop->total_time % USECS_PER_SEC) / USECS_PER_MSEC,
                        thr_loop->exec_total_time / USECS_PER_SEC, (thr_loop->exec_total_time % USECS_PER_SEC) / USECS_PER_MSEC);
            if(kind == RUN_KIND_BATCHINSERT)
            {
                printf("prepare time %lds %ldms\n", 
                            thr_loop->prepare_total_time / USECS_PER_SEC,
                            (thr_loop->prepare_total_time % USECS_PER_SEC) / USECS_PER_MSEC);
                printf("conn time %lds %ldms\n", 
                            thr_loop->exec_conn_time / USECS_PER_SEC,
                            (thr_loop->exec_conn_time % USECS_PER_SEC) / USECS_PER_MSEC);

                printf("dou time %lds %ldms\n", 
                            thr_loop->exec_total_dou_time / USECS_PER_SEC,
                            (thr_loop->exec_total_dou_time % USECS_PER_SEC) / USECS_PER_MSEC);
                
            }
        }
        if(test_mode == TESTMODE_FIX_TIME) {
            printf("\nThread %d line number %ld TPS:%ld\n\n", thr_loop->thread_id, thr_loop->tuple_num_fix_time
                        , thr_loop->tuple_num_fix_time / (thr_loop->exec_total_time / USECS_PER_SEC));
        }

        printf("\n");
        
    }
}

static void collect_thread(void)
{
    int loop = 0;
    for (loop = 0; loop < nthread; loop++) {
        ThreadData *thr_loop = &thread_handle[loop];

        all_exec_time += thr_loop->exec_total_time;
        all_conn_time += thr_loop->exec_conn_time;
        all_prepare_time += thr_loop->prepare_total_time;
        all_db_time = all_exec_time + all_conn_time + all_prepare_time;
        all_thread_time += thr_loop->total_time;
        all_tuples_fix_time += thr_loop->tuple_num_fix_time;
        all_produce_time += thr_loop->produce_data_time;
        all_dou_time += thr_loop->exec_total_dou_time;

        max_thread_time = thr_loop->total_time > max_thread_time ? thr_loop->total_time : max_thread_time;
    }
}


static void free_thread(void)
{
    int loop = 0;

    for (loop = 0; loop < nthread; loop++) {
        ThreadData *th_ptr = &thread_handle[loop];

        disconnect(&th_ptr->conn);
        if(th_ptr->paramValues)
            free(th_ptr->paramValues);
    }

    free(thread_handle);
}

static char* get_partition_table_sql(bool pt, bool tt, char *create_table_normal, int index)
{
    if(!pt && !tt)
        sprintf(create_table_normal, "CREATE TABLE IF NOT EXISTS %s_%d(id INT, profit INT, describ TEXT, t TIMESTAMP);", table_name, index);
    else if (tt) {
        sprintf(create_table_normal, "CREATE TIMESERIES TABLE IF NOT EXISTS %s_%d(t TIMESTAMP TSTIME, name TEXT TSTAG, fleet TEXT TSTAG, driver TEXT TSTAG, \
            model TEXT TSTAG, device_version TEXT TSTAG, load_capacity TEXT TSTAG, fuel_capacity TEXT TSTAG, nominal_fuel_consumption TEXT TSTAG, \
            latitude double precision, longitude double precision, elevation double precision,velocity double precision,heading double precision, \
            grade double precision, fuel_consumption double precision);", table_name, index);
    }
    else {
        char    *ptr = create_table_normal;

        sprintf(ptr, "CREATE TABLE IF NOT EXISTS %s_%d(id INT, profit INT, describ TEXT, t TIMESTAMP) PARTITION BY RANGE(t) SUBPARTITION BY HASH(describ) ( \n", table_name, index);
        ptr += strlen(ptr);
        sprintf(ptr, "partition p1 values less than('2030-12-08 00:00:00') (subpartition p1_1, subpartition p1_2, subpartition p1_3)");
        ptr += strlen(ptr);
        sprintf(ptr, ");");
    }
    
    return create_table_normal;
}

static void init_tables(void)
{
    PGconn  *cur_conn = NULL;
    char    **truncate_table_sql_array = NULL;
    char    **drop_table_sql_array = NULL;
    char    **create_table_sql_array = NULL;
    char    **table_name_array = NULL;
    int     table_number = 1;
    int     loop = 0;

    if(ntables  != 1 && (kind == RUN_KIND_NOMAL || kind == RUN_KIND_PREPARED
                                || kind == RUN_KIND_PREPARED_WITHXACT))
        table_number =  ntables;

    table_name_array = (char**)malloc(table_number * sizeof(char*));
    truncate_table_sql_array = (char**)malloc(table_number * sizeof(char*));
    drop_table_sql_array = (char**)malloc(table_number * sizeof(char*));
    create_table_sql_array = (char**)malloc(table_number * sizeof(char*));

    for(loop = 0; loop < table_number; loop++)
    {
        table_name_array[loop] = (char*)malloc(NAMEDATALEN);
        truncate_table_sql_array[loop] = (char*)malloc(SQL_MAX_LENGTH);
        drop_table_sql_array[loop] = (char*)malloc(SQL_MAX_LENGTH);
        create_table_sql_array[loop] = (char*)malloc(SQL_MAX_LENGTH);
        
        sprintf(table_name_array[loop], "%s_%d", table_name, loop);
        sprintf(truncate_table_sql_array[loop], "TRUNCATE TABLE %s", table_name_array[loop]);
        if(timeseries_table)
            sprintf(drop_table_sql_array[loop], "DROP TIMESERIES TABLE IF EXISTS %s", table_name_array[loop]);
        else
            sprintf(drop_table_sql_array[loop], "DROP TABLE IF EXISTS %s", table_name_array[loop]);
        get_partition_table_sql(partition_table, timeseries_table, create_table_sql_array[loop], loop);
/*printf("table_name_array[%d]=%s \n", loop, table_name_array[loop]);
printf("truncate_table_sql_array[%d]=%s \n", loop, truncate_table_sql_array[loop]);
printf("drop_table_sql_array[%d]=%s \n", loop, drop_table_sql_array[loop]);
printf("create_table_sql_array[%d]=%s \n", loop, create_table_sql_array[loop]);*/
    }

    cur_conn = doConnect();
    if(target_rel_mo == 2)
    {
        for(loop = 0; loop < table_number; loop++)
        {
            executeCommand(cur_conn, drop_table_sql_array[loop]);
            executeCommand(cur_conn, create_table_sql_array[loop]);
        }
    }
    else if( target_rel_mo == 1)
    {
        for(loop = 0; loop < table_number; loop++)
        {
            check_table_exists(cur_conn, table_name_array[loop]);
            executeCommand(cur_conn, truncate_table_sql_array[loop]);
        }
    }
    else
    {
        if(!(kind == RUN_KIND_QUERY || kind == RUN_KIND_PREPARE_QUERY)) {
            for(loop = 0; loop < table_number; loop++)
            {
                check_table_exists(cur_conn, table_name_array[loop]);
            }
        }
    }
    disconnect(&cur_conn);
}

static void init(void)
{
    init_tables();
    init_thread();
}

void help(void)
{
    printf("my_test_tool help\n");
    printf("my_test_tool option:\n");
    printf("[option]\n");
    printf("\n[with arg]\n");
    printf("-j parallel clients\n");
    printf("-s tuples we insert totally\n");
    printf("-k 1:nomal insert 2:prepare insert\n");
    printf("   3:prepare insert with xact\n");
    printf("   4:batch insert INSERT INTO VALUES(),()...\n");
    printf("   5:like 4 without bind\n");
    printf("   6:INSERT INTO VALUES();INSERT INTO VALUES(); with nobind\n");
    printf("   7:COPY batch\n");
    printf("   8:gauss batch insert API\n");
    printf("   (1 default)\n");
    printf("-S use per xact every batch if k=4 or 5 or 6 or 7 or 8\n");
    printf("-B batch per xact of k=3,4,5,6,7,8\n");
    printf("-n test tables if k=1,2,3\n");
    printf("-r test relname\n");
    printf("-i 0: use table direct 1:truncate 2: recreate (default 2)\n");
    printf("-M 0: fix lines mode 1:fix time mode (default 0)\n");
    printf("-T run time by second\n");
    printf("-H host name\n");
    printf("-P connect port\n");
    printf("-U connect user\n");
    printf("-d connect database name\n");
    printf("-W connect password\n");

    printf("\n[without arg]\n");
    printf("-b enable bypass (default off)\n");
    printf("-f ensable fast extend (default off)\n");
    printf("-C ensable plan recycle (default off)\n");
    printf("-t use timeseries table (default false)\n");
    printf("-p use partation table (defause false)\n");
    printf("-R replan for every batch if k=4 (defause false)\n");
    printf("-m enable batchinsert to muti if k=4 (default off)\n");
    printf("-v out detail\n");
    printf("-h print this help\n");
    printf("my_test_tool  -j20 -k4 -B 2000  -M 1 -T 10 -t\n");
    printf("my_test_tool  -j1 -k4 -B 2000 -s 10 -t\n");

    exit(0);
}

int main(int argc, char** argv)
{
    char    c = 0;
    int     optindex =0;

    static struct option long_options[] = {
        {NULL, 0, NULL, 0}};

    while ((c = getopt_long(argc, argv, "tj:s:i:ph?k:bmvr:B:SQAM:T:H:fP:U:d:W:RCn:", long_options, &optindex)) != -1)
    {
        switch (c) {
            case 'j':
                nthread = atoi(optarg);
                if (nthread <= 0) {
                    fprintf(stderr, "invalid number of threads(-j): %d\n", nthread);
                    exit(1);
                }
                break;
            case 'n':
                ntables = atoi(optarg);
                if (nthread <= 0) {
                    fprintf(stderr, "invalid number of taables(-n): %d\n", nthread);
                    exit(1);
                }
                break;
            case 's':
                total_round = atoi(optarg);
                if (total_round <= 0) {
                    fprintf(stderr, "invalid number of scale(-s): %d\n", total_round);
                    exit(1);
                }
                break;
            case 'i':
                do_init = true;
                target_rel_mo = atoi(optarg);
                if (target_rel_mo != 0 && target_rel_mo != 1 && target_rel_mo != 2) {
                    fprintf(stderr, "invalid number of target_rel_mo(-i): %d\n", target_rel_mo);
                    exit(1);
                }
                break;
            case 'T':
                run_seconds = atoi(optarg);
                total_round = -1;
                if (run_seconds > 10000) {
                    fprintf(stderr, "invalid number of run_seconds(-T): %s\n", optarg);
                    exit(1);
                }
                break;
            case 'H':
                hostname = strdup(optarg);
                break;
            case 'P':
                conf_port = strdup(optarg);
                break;
            case 'U':
                conf_user_name = strdup(optarg);
                break;
            case 'd':
                conf_database = strdup(optarg);
                break;
            case 'W':
                conf_password = strdup(optarg);
                break;
            case 'b':
                enable_opfusion = true;
                break;
            case 'C':
                enable_plan_recycle = true;
                break;
            case 'f':
                fast_extend = true;
                break;
            case 'm':
                enable_batchinsert_to_muti = true;
                break;
            case 'M':
                test_mode = atoi(optarg);
                if(test_mode != TESTMODE_FIX_LINES && test_mode != TESTMODE_FIX_TIME)
                {
                    fprintf(stderr, "invalid number of test_mode(-M): %d !!!!!!!!\n", test_mode);
                    exit(1);
                }
                break;
                
            case 'B':
                batch_per_xact = atoi(optarg);
                if(batch_per_xact <= 0)
                {
                    fprintf(stderr, "invalid number of batch(-B): %d !!!!!!!!\n", batch_per_xact);
                    exit(1);
                }
                break;
            case 'p':
                partition_table = true;
                table_name = table_name_partition;
                break;
            case 't':
                timeseries_table = true;
                table_name = table_name_ts;
                break;
            case 'R':
                replan_every_batch = true;
                break;
            case 'k':
                kind = atoi(optarg);
                if(kind <= 0 || kind >= RUN_KIND_MAX)
                {
                    fprintf(stderr, "invalid number of kind(-k): %d\n", kind);
                    exit(1);
                }
                break;
            case 'v':
                verbose = true;
                break;
            case 'h':
            case '?':
                help();
                break;
            case 'r':
                if(!optarg)
                {
                    fprintf(stderr, "invalid arg of relname(-r): NULL\n");
                    exit(1);
                }
                memset(relname, 0, SQL_MAX_LENGTH);
                strcpy(relname, optarg);
                table_name = relname;
                break;
            case 'S':
                short_conn = true;
                break;
            case 'Q':
                dir_query = true;
                break;
            case 'A':
                auto_test = true;
                break;
            default:
                help();
                break;
        }
        get_arg = true;
    }

    get_conn_str();

    if(dir_query)
    {
        handle_query();
        exit(1);
    }

    if(kind == RUN_KIND_QUERY)
    {
        get_conn_str1();
        target_rel_mo = 0;
        //handle_auto_test();
        //exit(1);
    }

    if(kind == RUN_KIND_PREPARE_QUERY) {
        get_conn_str2();
        target_rel_mo = 0;
    }

    if(!table_name)
        table_name = table_name_nomal;

    if(!get_arg)
    {
        help();
    }

    if(timeseries_table && partition_table) {
        printf("Can not special -p and -t at same time");
        exit(1);
    }
    if(timeseries_table && target_rel_mo == 1) {
        printf("Not support truncate mode in timeseries table mode");
        exit(1);
    }

    if(test_mode == TESTMODE_FIX_TIME && run_seconds == 0) {
        printf("You need run_seconds(-T) for test_mode=1");
        exit(1);
    }

    if(total_round == -1)
    {
        avg_round = -1;
        first_round = -1;
    }
    else 
    {
        avg_round = total_round / nthread;
        first_round = total_round - (avg_round * (nthread - 1));
    }


    init();
    gettimeofday(&start_time, NULL);

    run_work = true;

    count_run_time();

    join_thread();

    collect_thread();
    if(verbose)
        dump_thread();
    gettimeofday(&stop_time, NULL);

    free_thread();

    
    real_time = get_time_diff_us(start_time, stop_time);
    //if(test_mode == TESTMODE_FIX_LINES) {
        printf("Thread ALL:\nRun time %lds %ldms\n", 
                        real_time / USECS_PER_SEC, (real_time % USECS_PER_SEC) / USECS_PER_MSEC);
        printf("Max thread time %lds %ldms\n", 
                        max_thread_time / USECS_PER_SEC, (max_thread_time % USECS_PER_SEC) / USECS_PER_MSEC);
        printf("All thread time %lds %ldms, doutime %lds %ldus produce time %lds %ldus, db time %lds %ldms, conn time %lds %ldms, prepare time %lds %ldms, exec time %lds %ldms\n\n", 
                        all_thread_time / USECS_PER_SEC, (all_thread_time % USECS_PER_SEC) / USECS_PER_MSEC,
                        all_dou_time / USECS_PER_SEC, (all_dou_time % USECS_PER_SEC) / USECS_PER_MSEC,
                        all_produce_time / USECS_PER_SEC, (all_produce_time % USECS_PER_SEC) / USECS_PER_MSEC,
                        all_db_time / USECS_PER_SEC, (all_db_time % USECS_PER_SEC) / USECS_PER_MSEC,
                        all_conn_time / USECS_PER_SEC, (all_conn_time % USECS_PER_SEC) / USECS_PER_MSEC,
                        all_prepare_time / USECS_PER_SEC, (all_prepare_time % USECS_PER_SEC) / USECS_PER_MSEC,
                        all_exec_time / USECS_PER_SEC, (all_exec_time % USECS_PER_SEC) / USECS_PER_MSEC);
    //} else if(test_mode == TESTMODE_FIX_TIME) {
        printf("All thread line number %ld TPS:%ld\n\n", all_tuples_fix_time, all_tuples_fix_time / (real_time / USECS_PER_SEC));
    //}

}
